import collections
import datetime
import math
import time
import traceback
import json
import subprocess
import tempfile
import logging
import toml
import pandas as pd
import numpy as np

from absl import app, flags
import pytz

import coin.base.database.util as db_util
from coin.proto.coin_market_query_pb2 import (
    ProductKlineElement,
    ProductKline,
)
from coin.support.market_quotes.logic.market_status_util import query_history_kline_list
from coin.exchange.kr_rest.product.product_impl import generate_product_from_str2
from xunkemgmt_client.market.market_quotes.database.mysql_importer import (
  MysqlCoinInfoImporter)
from xunkemgmt_client.client.util.query_util import query_exchange_apis
from xunkemgmt_client.support.accounting.database.db_output import DBOutputKline
from coin.support.pta.logic.pta_stats_calculator import (calculate_turnover,
                                                         calculate_turnover_in_usd,
                                                         get_accounting_currency)
from xunkemgmt_client.market.feed_stats_v2.database.connection.coin2_database import Session
from xunkemgmt_client.market.feed_stats_v2.model.feed_stats import (
    FeedStatsIndex, TradeStats)
from coin.base.query_util import (
    query_exchange_rates,
    query_klines,
    query_exchange_tickers)
from coin.support.accounting.logic.query_util import (
    filter_invalid_request_and_result,
    ExchangeApiRequest)
from coin.support.market_quotes.logic.query_util import (
    KlineRequest,
    get_kline_requests,
    drop_duplicate_kline_request,
    select_kline_request)
from coin.support.accounting.app.account_info_collector import (
    _load_proxy_config,
    _get_proxy_config_file)

FLAGS = flags.FLAGS

NO_MULTITHREAD_MARKET = [('Spot', 'Binance'), ('Spot', 'Huobi')]

KLINE_RATE_LIMIT = {
    'Binance': 9,
    'Gdax': 2,
    'Okex': 9,
    'Huobi': 9,
    'Upbit': 9,
}  # max kline query per second

MARKET_TYPE_EXCHANGE_PY_LIST = []


def separate_request_by_exchange(request_list):
  request_collection_summary = collections.Counter([r.exchange for r in request_list])
  max_group_required = [
      math.ceil(v / KLINE_RATE_LIMIT[k]) for k,
      v in request_collection_summary.items() if k in KLINE_RATE_LIMIT
  ]
  if len(max_group_required) == 0:
    return [request_list]
  max_group = max(max_group_required)
  request_groups = []
  for i in range(max_group):
    if i == max_group - 1:
      request_group = request_list
    else:
      request_group = []
      for exchange in request_collection_summary:
        choose_number = math.ceil(request_collection_summary[exchange] / max_group)
        chooose_requests = [r for r in request_list if r.exchange == exchange][:choose_number]
        request_group.extend(chooose_requests)
        request_list = [r for r in request_list if r not in chooose_requests]
    request_groups.append(request_group)
  return request_groups


def _check_kline_symbol_exist(symbol_groups_universe,
                              market_type,
                              exchange,
                              api_version,
                              symbol):
  mea = f'{market_type.lower()}_{exchange.lower()}_{api_version.lower()}'
  symbol_list = [symbol for symbol_group in
                 [*(symbol_groups_universe[mea].values())]
                 for symbol in symbol_group]
  if market_type == 'Options' and exchange == 'Deribit':
    symbol_list = [symbol.replace('.ALL', '') for symbol in symbol_list]
    symbol = symbol.strip().split('.')[:3]
    symbol = '.'.join(symbol)
  return symbol in symbol_list


def kline_requests_to_driver_dict(kline_requests, kline_length=50):
  mea_config_dict = {}
  exchange_api_df = query_exchange_apis().set_index(['exchange_api_id'])
  symbol_group_file = 'data/coin2/feed/symbol_groups.toml'
  mea_symbol_dict = {}
  with open(symbol_group_file) as f:
    symbol_groups_universe = toml.load(f)
  for kline_request in kline_requests:
    try:
      assert kline_request.market_type is not None
      assert kline_request.exchange is not None
      assert kline_request.source_exchange_api_id is not None
      assert kline_request.symbol_contract_type is not None
      market_type = kline_request.market_type
      exchange = kline_request.exchange
      api_version = exchange_api_df.loc[kline_request.source_exchange_api_id, 'api_version']
      norm_symbol = kline_request.symbol_contract_type
      mea = f'{market_type}.{exchange}.{api_version}'
      assert _check_kline_symbol_exist(
        symbol_groups_universe, market_type,
        exchange, api_version, norm_symbol), norm_symbol
    except Exception as e:
      logging.error(f'failed symbol: {norm_symbol} for {mea}', type(e), e)
      continue
    if mea in mea_symbol_dict:
      mea_symbol_dict[mea].append(norm_symbol)
    else:
      mea_symbol_dict[mea] = [norm_symbol]
  for mea, symbol_list in mea_symbol_dict.items():
    market_type, exchange = mea.strip().split('.')[:2]
    http_proxy = None
    proxy_request = KlineRequest(market_type=market_type, exchange=exchange)
    config_file = _get_proxy_config_file(FLAGS.proxy_info_file, proxy_request) \
      if FLAGS.proxy_info_file is not None else None
    if config_file:
      http_proxy = _load_proxy_config(config_file)['http']
    mea_config_dict[mea] = {
        'channels': ['kline_rest_1hour'],
        'products': {'norms': symbol_list},
        'websocket': {
          'urls': {
            'kline_rest':
              {
                'http_proxy': http_proxy
              }
          }
        }
    }
    if market_type == 'Options':
      mea_config_dict[mea]['products'].update({"query_and_update_symbols": True})
  common_config = {
      "type": "FEED_SYSTEM_TYPE_LIVE",
      "websocket": {
          "num_workers": 1,
          "subscribe_one_time": True,
          "kline_length": str(kline_length)
      }
  }
  driver = {'feed': {'common': common_config, 'exchanges': mea_config_dict}}
  return driver


def _query_cc_klines(request_list, trading_date, out_dir=None):
  if len(request_list) == 0:
    return
  cur_dt = datetime.datetime.utcnow().replace(tzinfo=pytz.UTC)
  if trading_date is None:
    trading_date = cur_dt - datetime.timedelta(1)
  start_dt = datetime.datetime.combine(
      trading_date, datetime.datetime.min.time()).replace(tzinfo=pytz.UTC)
  time_range_length = int((cur_dt - start_dt).total_seconds() / 3600) + 24
  dt_str = trading_date.strftime('%Y%m%d')
  out_dir = tempfile.mkdtemp(dir='/tmp', prefix=f'kline_{dt_str}_') \
    if out_dir is None else out_dir
  driver_dict = kline_requests_to_driver_dict(request_list, time_range_length)
  out_driver_json = tempfile.mkstemp(
      dir='/tmp', prefix=f'driver_config.', suffix='.json')[-1]
  with open(out_driver_json, 'w') as fwrite:
    json.dump(driver_dict, fwrite, indent=2, sort_keys=True)
  cmd = 'source ${HOME}/miniconda3/etc/profile.d/conda.sh' + ' && ' + \
        'conda activate coin2_motion_env_migration' + ' && ' + \
        'export GLOG_logtostderr=1' + ' && cd .. && ' + \
        '../coin_binary/coin2-bazel-bin-3.10/cc/appcoin2/support/feed/kline_csv ' + \
        '--date %s --driver %s --out-dir %s > %s 2>&1' % (
            dt_str, out_driver_json, out_dir, f'./kline_info_collector.{dt_str}')
  subprocess.run(cmd, shell=True, check=True)
  return out_dir


def get_cc_kline_requests_list(trading_date, kline_info_df):
  result_list = []
  exchange_api_df = query_exchange_apis().set_index(
      ['market_type', 'exchange', 'api_version'])
  for _, row in kline_info_df.iterrows():
    market_type, exchange, api_version = row.mea.split('.')
    symbol_contract_type = row.symbol
    kline_period = 'KLINE_INTERVAL_1HOUR'
    if (market_type, exchange, api_version) in exchange_api_df.index:
      source_exchange_api_id = exchange_api_df.loc[(
          market_type, exchange, api_version), 'exchange_api_id']
    else:
      raise ValueError("Not in exchange_api (%s, %s, %s)" % (market_type, exchange, api_version))
    result_list.append(KlineRequest(
        market_type, exchange, symbol_contract_type, kline_period, source_exchange_api_id, trading_date, time.time()))
  result_list = drop_duplicate_kline_request(result_list)
  return result_list


def collect_cc_klines(kline_request_list, start_time, end_time):
  if len(kline_request_list) == 0:
    return [], []
  trading_date = start_time.date()
  kline_data_dir = _query_cc_klines(kline_request_list, trading_date)
  trading_date_str = trading_date.strftime('%Y%m%d')
  kline_info_df = pd.read_csv(f'{kline_data_dir}/{trading_date_str}.csv')
  kline_request_list = get_cc_kline_requests_list(trading_date, kline_info_df)
  mea_symbol_list = [(req.market_type, req.exchange, req.symbol_contract_type)
                      for req in kline_request_list]
  kline_response_list = []
  for market_type, exchange, symbol in mea_symbol_list:
    product_df = kline_info_df[(kline_info_df['mea'].str.match(
        f'{market_type}.{exchange}')) & (kline_info_df['symbol'] == symbol)]
    product_df = product_df.replace(np.nan, None)
    market_type, exchange, api_version = product_df.iloc[0].mea.split('.')
    symbol_contract_type = product_df.iloc[0].symbol
    product = generate_product_from_str2(
        market_type, exchange, api_version, symbol_contract_type, start_time)
    try:
      product_klines = []
      for _, row in product_df.iterrows():
        assert isinstance(row.open, float)
        assert isinstance(row.high, float)
        assert isinstance(row.low, float)
        assert isinstance(row.close, float)
        assert pd.notnull(row.volume)
        product_kline = ProductKlineElement(
            open=row.open,
            high=row.high,
            low=row.low,
            close=row.close,
            volume=row.volume,
            turnover=row.turnover,
            kline_timestamp=int(datetime.datetime.strptime(
                row.open_timestamp, '%Y-%m-%dT%H:%M:%S').replace(tzinfo=pytz.UTC).timestamp() * 1e+9))
        product_klines.append(product_kline)
      klines = ProductKline(symbol=product.symbol,
                            native_symbol=product.native_symbol,
                            exchange=exchange,
                            market_type=market_type,
                            kline_period='KLINE_INTERVAL_1HOUR',
                            klines=product_klines)
      kline_result = DBOutputKline(
          query_timestamp=datetime.datetime.utcnow().timestamp(), klines=klines)
    except Exception as e:
      logging.error('Fail to parse kline.', type(e), e)
      kline_result = None
      logging.error(traceback.format_exc())
    finally:
      kline_response_list.append(kline_result)
  kline_request_list, kline_response_list = \
    filter_invalid_request_and_result(kline_request_list, kline_response_list)
  return kline_request_list, kline_response_list


def collect_klines(kline_request_list, start_time, end_time):
  if len(kline_request_list) == 0:
    return [], []
  kline_request_list_ordered = []
  kline_response_list = []
  kline_request_multithread_list = []
  for request in kline_request_list:
    if (request.market_type, request.exchange) in NO_MULTITHREAD_MARKET:
      kline_response_list.extend(query_history_kline_list(
          [request], start_time, end_time, no_throw=True))
      kline_request_list_ordered.append(request)
      time.sleep(0.1)
    else:
      kline_request_multithread_list.append(request)
  kline_request_groups = separate_request_by_exchange(kline_request_multithread_list)
  for kline_request_group in kline_request_groups:
    kline_response_list.extend(query_history_kline_list(
        kline_request_group, start_time, end_time, no_throw=True))
    kline_request_list_ordered.extend(kline_request_group)
    time.sleep(1)
  kline_request_list = kline_request_list_ordered
  kline_request_list, kline_response_list = \
    filter_invalid_request_and_result(kline_request_list, kline_response_list)
  return kline_request_list, kline_response_list


def collect_klines_from_feed_stats(kline_request_list, start_time, end_time):
  if len(kline_request_list) == 0:
    return [], []
  exchange_api_df = query_exchange_apis()
  trading_date_str = datetime.datetime.strftime(start_time.date(), '%Y-%m-%d')
  strat_time_str = datetime.datetime.strftime(start_time, '%Y-%m-%d %H:%M:%S')
  feed_stats = Session.query(FeedStatsIndex.feed_stats_id, FeedStatsIndex.trading_date,
                                        FeedStatsIndex.exchange_api_id, FeedStatsIndex.symbol).filter(
                                        FeedStatsIndex.trading_date == trading_date_str).all()
  trade_stats = Session.query(TradeStats.feed_stats_id, TradeStats.total_volume, TradeStats.vwap).filter(
                                        TradeStats.time_first_trade >= strat_time_str).all()
  if not feed_stats and not trade_stats:
    print(f'No data for trading date {trading_date_str} yet')
    return [], []
  feed_stats_df = pd.merge(pd.DataFrame(feed_stats), pd.DataFrame(trade_stats), on='feed_stats_id')
  feed_stats_df = pd.merge(feed_stats_df, exchange_api_df, on='exchange_api_id')
  feed_stats_df = feed_stats_df.groupby(
      ['trading_date', 'market_type', 'exchange', 'symbol']).tail(1)
  kline_response_list = []

  for request in kline_request_list:
    try:
      assert request.symbol_contract_type == 'TOTAL'
      df = feed_stats_df[(feed_stats_df['market_type'] == request.market_type)
                          & (feed_stats_df['exchange'] == request.exchange)]
      assert len(df) > 0, 'exchange not supported'
      total_turnover_in_usd = 0
      for _, product in df.iterrows():
        kline_ts = start_time.replace(tzinfo=pytz.UTC).timestamp() * 1e+9
        try:
          product_kline = ProductKlineElement(
              volume=product['total_volume'], close=product['vwap'], kline_timestamp=int(kline_ts))
          product_request = KlineRequest(market_type=product['market_type'], exchange=product['exchange'],
                                         symbol_contract_type=product['symbol'], trading_date=request.trading_date)
          product_kline = _regenerate_kline(product_request, product_kline)
          total_turnover_in_usd += product_kline.turnover_in_usd
        except Exception as e:
          logging.error('Fail to calculate turnover in usd of symbol.',
                        product['market_type'], product['exchange'], product['symbol'], type(e), e)
          continue
      assert total_turnover_in_usd != 0
      kline_dict = ProductKlineElement(
          turnover_in_usd=total_turnover_in_usd,
          kline_timestamp=int(kline_ts))
      total_kline = ProductKline(
                          symbol='TOTAL',
                          native_symbol='TOTAL',
                          market_type=request.market_type,
                          exchange=request.exchange,
                          kline_period='KLINE_INTERVAL_1DAY',
                          klines=[kline_dict])
      kline_result = DBOutputKline(
          query_timestamp=datetime.datetime.utcnow().timestamp(), klines=total_kline)
    except Exception as e:
      logging.error('Fail to query kline.', request.market_type,
                    request.exchange, request.symbol_contract_type, type(e), e)
      kline_result = None
    finally:
      kline_response_list.append(kline_result)
  kline_request_list, kline_response_list = \
    filter_invalid_request_and_result(kline_request_list, kline_response_list)
  return kline_request_list, kline_response_list


def _get_exchange_api_id(fx_ticker_request):
  data = query_exchange_apis()
  exchange_api_id_df = data[(data['exchange'] == fx_ticker_request.exchange)
                            & (data['market_type'] == fx_ticker_request.market_type)]
  return exchange_api_id_df['exchange_api_id'].iloc[0]


def _parse_fx_ticker_to_kline(fx_ticker_request, fx_ticker, trading_date, query_ts):
  fx_kline_request_list = []
  fx_kline_response_list = []
  exchange_api_id = _get_exchange_api_id(fx_ticker_request)
  for each_ticker in fx_ticker.each_ticker:
    fx_kline_request_list.append(
        KlineRequest(fx_ticker_request.market_type,
                     fx_ticker_request.exchange,
                     each_ticker.symbol,
                     'KLINE_INTERVAL_1DAY',
                     exchange_api_id,
                     trading_date,
                     time.time()))
    product_kline = ProductKlineElement(kline_timestamp=query_ts,
                                        close=float(each_ticker.last))
    klines = ProductKline(symbol=each_ticker.symbol,
                          exchange=fx_ticker.exchange,
                          market_type=fx_ticker_request.market_type,
                          kline_period='KLINE_INTERVAL_1DAY',
                          klines=[product_kline])
    fx_kline_response_list.append(
        DBOutputKline(query_timestamp=datetime.datetime.utcnow().timestamp(), klines=klines))
  return fx_kline_request_list, fx_kline_response_list


def convert_ticker_to_kline(trading_date, kline_config_path):
  query_time = datetime.datetime.combine(trading_date, datetime.datetime.max.time()).timestamp()
  query_ts = int(query_time * 10**9)
  fx_ticker_request = ExchangeApiRequest(market_type='Info',
                                         exchange='ExchangeRates',
                                         api_version=None)
  fx_ticker = query_exchange_tickers(query_ts=query_ts,
                                     market_type='Info',
                                     exchange='ExchangeRates')
  fx_kline_request_list, fx_kline_response_list = \
      _parse_fx_ticker_to_kline(fx_ticker_request, fx_ticker, trading_date, query_ts)
  insert_product_kline_db(fx_kline_request_list, fx_kline_response_list, kline_config_path)


def _regenerate_kline(request, kline):
  kline_new = ProductKlineElement()
  kline_new.CopyFrom(kline)
  assert kline.HasField('volume') and kline.HasField(
      'close') and kline.HasField('kline_timestamp'), kline
  query_dt = datetime.datetime.utcfromtimestamp(kline.kline_timestamp // 10**9)
  if not kline.HasField('turnover'):
    turnover = calculate_turnover(market_type=request.market_type,
                                  exchange=request.exchange,
                                  symbol=request.symbol_contract_type,
                                  volume=kline.volume,
                                  price=kline.close,
                                  current_datetime=query_dt)
    if turnover is not None:
      kline_new.turnover = turnover
  else:
    turnover = kline.turnover
  if not kline.HasField('turnover_in_usd'):
    accounting_currency = get_accounting_currency(request.market_type,
                                                  request.exchange,
                                                  request.symbol_contract_type,
                                                  query_dt)
    price_in_usd = query_exchange_rates([accounting_currency],
                                       'USD',
                                        request.trading_date)[0]
    turnover_in_usd = calculate_turnover_in_usd(market_type=request.market_type,
                                                exchange=request.exchange,
                                                symbol=request.symbol_contract_type,
                                                volume=kline.volume,
                                                turnover=turnover,
                                                turnover_price_in_usd=price_in_usd,
                                                current_datetime=query_dt)
    if turnover_in_usd is not None:
      kline_new.turnover_in_usd = turnover_in_usd
  return kline_new


def aggregate_kline(request_list, response_list):
  result_list = []
  assert len(request_list) == len(response_list)
  for request, response in zip(request_list, response_list):
    assert isinstance(response, DBOutputKline), response
    product_kline = response.klines
    assert isinstance(product_kline, ProductKline), product_kline
    try:
      assert len(product_kline.klines) > 0, product_kline
      new_klines = []
      for each_kline in product_kline.klines:
        each_kline = _regenerate_kline(request, each_kline)
        new_klines.append(each_kline)
      trading_datetime = datetime.datetime.combine(request.trading_date,
                                                   datetime.datetime.min.time())
      assert all([kline.HasField('volume') for kline in new_klines])
      volume = sum([x.volume for x in new_klines])
      assert all([kline.HasField('turnover') for kline in new_klines])
      turnover = sum([x.turnover for x in new_klines])
      has_turnover_in_usd = all([kline.HasField('turnover_in_usd') for kline in new_klines])
      turnover_in_usd = sum([x.turnover_in_usd for x in new_klines]) if \
                        has_turnover_in_usd else None
      product_kline_element = ProductKlineElement(
          open=new_klines[0].open,
          high=max([x.high for x in new_klines]),
          low=min([x.low for x in new_klines]),
          close=new_klines[-1].close,
          volume=volume,
          turnover=turnover,
          turnover_in_usd=turnover_in_usd,
          kline_timestamp=int(trading_datetime.replace(tzinfo=pytz.UTC).timestamp() * 1e+9))
      result = DBOutputKline(query_timestamp=response.query_timestamp,
                             klines=ProductKline(symbol=product_kline.symbol,
                                                 native_symbol=product_kline.native_symbol,
                                                 market_type=product_kline.market_type,
                                                 exchange=product_kline.exchange,
                                                 kline_period='KLINE_INTERVAL_1DAY',
                                                 klines=[product_kline_element]))
    except Exception as e:
      logging.error('Fail to aggregate kline.', request, type(e), e)
      result = None
      logging.error(traceback.format_exc())
    result_list.append(result)
  return result_list


def convert_to_1d_kline(kline_request_list, trading_date, config_path):
  request_list = [
      request for request in kline_request_list
      if request.kline_period == 'KLINE_INTERVAL_1HOUR' or
      (request.kline_period == 'KLINE_INTERVAL_1DAY' and request.exchange == 'Gopax')
  ]
  logging.info('start query kline from db')
  kline_output = query_klines(start_date=trading_date,
                              end_date=trading_date,
                              period='KLINE_INTERVAL_1HOUR',
                              as_proto=True)
  kline_output_dict = {
      (kline.market_type, kline.exchange, kline.relative_symbol): kline for kline in kline_output}
  kline_output_list = [kline_output_dict.get(
      (request.market_type, request.exchange, request.symbol_contract_type), None) for request in request_list]
  request_list, kline_output_list = filter_invalid_request_and_result(request_list, kline_output_list)
  kline_output_list = [
      DBOutputKline(query_timestamp=datetime.datetime.utcnow().timestamp(), klines=output)
      if output is not None else None for output in kline_output_list
  ]
  logging.info('start aggregate kline')
  response_list = aggregate_kline(request_list, kline_output_list)
  converted_request_list = []
  for kline_request in request_list:
    kline_request_dict = kline_request._asdict()
    kline_request_dict['kline_period'] = 'KLINE_INTERVAL_1DAY'
    kline_request_dict['query_timestamp'] = time.time()
    converted_request_list.append(KlineRequest(**kline_request_dict))
  converted_request_list, response_list = \
    filter_invalid_request_and_result(converted_request_list, response_list)
  logging.info('start insert converted kline')
  insert_product_kline_db(converted_request_list, response_list, config_path)


def insert_product_kline_db(kline_request_list, kline_response_list, config_path):
  assert len(kline_request_list) == len(kline_response_list)
  mysql_config = db_util.read_db_config(config_path)
  importer = MysqlCoinInfoImporter(db_config=mysql_config)
  importer.insert_klines(kline_request_list, kline_response_list)


def main(_):
  max_date = datetime.datetime.utcnow().date() - datetime.timedelta(days=1)
  if FLAGS.start_date is None and FLAGS.end_date is None:
    start_date = max_date
    end_date = start_date
  else:
    start_date = datetime.datetime.strptime(FLAGS.start_date, '%Y%m%d').date()
    end_date = datetime.datetime.strptime(FLAGS.end_date, '%Y%m%d').date()
  assert start_date <= end_date <= max_date, (start_date, end_date, max_date)
  trading_date = end_date
  while trading_date >= start_date:
    kline_request_list = get_kline_requests(trading_date, FLAGS.kline_symbols_list)
    kline_request_list = select_kline_request(kline_request_list,
                                              market_type=FLAGS.market_type,
                                              exchange=FLAGS.exchange,
                                              kline_period=FLAGS.kline_period)
    assert len(kline_request_list) > 0
    logging.info(trading_date)
    start_time = datetime.datetime.combine(
        trading_date, datetime.time.min).replace(tzinfo=pytz.UTC)
    end_time = datetime.datetime.combine(
        trading_date, datetime.time.max).replace(tzinfo=pytz.UTC)
    logging.info('start convert ticker to kline')
    convert_ticker_to_kline(trading_date, FLAGS.mysql_config)
    exchange_request_list = [req for req in kline_request_list
                              if req.symbol_contract_type == 'TOTAL']
    kline_request_list = list(set(kline_request_list) - set(exchange_request_list))
    logging.info('start collect kline from feed stats')
    feed_stats_request_list, feed_stats_response_list = collect_klines_from_feed_stats(
        exchange_request_list, start_time, end_time)
    logging.info('start collect kline from exchange')
    if FLAGS.legacy:
      kline_request_list, kline_response_list = collect_klines(
          kline_request_list, start_time, end_time)
    else:
      kline_py_request_list = [kline_request for kline_request in kline_request_list
                               if (kline_request.market_type, kline_request.exchange) in MARKET_TYPE_EXCHANGE_PY_LIST]
      kline_cc_request_list = list(set(kline_request_list) - set(kline_py_request_list))
      py_request_list, py_response_list = collect_klines(
          kline_py_request_list, start_time, end_time)
      cc_request_list, cc_response_list = collect_cc_klines(
          kline_cc_request_list, start_time, end_time)
      kline_request_list = py_request_list + cc_request_list
      kline_response_list = py_response_list + cc_response_list
    insert_product_kline_db(kline_request_list + feed_stats_request_list,
                            kline_response_list + feed_stats_response_list, FLAGS.mysql_config)
    logging.info('start convert kline')
    convert_to_1d_kline(kline_request_list, trading_date, FLAGS.mysql_config)
    trading_date -= datetime.timedelta(1)


if __name__ == '__main__':
  flags.DEFINE_string('kline_symbols_list',
                      '../../coin_deploy/support_info/kline_config/kline_symbols_list.csv',
                      'kline_symbols_list')
  flags.DEFINE_string('start_date', None, '%Y%m%d')
  flags.DEFINE_string('end_date', None, '%Y%m%d')
  flags.DEFINE_string('exchange', None, 'exchange')
  flags.DEFINE_string('market_type', None, 'market_type')
  flags.DEFINE_string('kline_period', None, 'kline_period')
  flags.DEFINE_string('mysql_config', None, 'path of database config.')
  flags.DEFINE_string('proxy_info_file',
                      '../../coin_deploy/support_accounting/proxy/market_proxy_info.toml',
                      'proxy_info_file')
  flags.DEFINE_boolean('legacy', False, 'py or cc collector')
  app.run(main)
